package sql

import org.apache.spark.SparkConf
import org.apache.spark.sql.expressions.Aggregator
import org.apache.spark.sql.{Encoder, Encoders, SparkSession, functions}

import scala.collection.mutable
import scala.collection.mutable.ListBuffer


object E2_TopN {


  """
    |需求： 各区域热门商品Top3
    |
    |我们需要计算各个区域前三大热门商品（这里的热门商品是从点击量的维度来看），并备注上每个商品在主要城市中的分布比例，超过两个城市用其他显示。
    |最终结果类似如下形式：
    |
    |地区	商品名称	点击次数	城市备注
    |华北	商品 A	100000	北京 21.2%，天津 13.2%，其他 65.6%
    |华北	商品 P	80200	北京 63.0%，太原 10%，其他 27.0%
    |华北	商品 M	40000	北京 63.0%，太原 10%，其他 27.0%
    |东北	商品 J	92000	大连 28%，辽宁 17.0%，其他 55.0%
    |""".stripMargin

  def main(args: Array[String]): Unit = {
    topN
  }


  """
    |
    |首先将用户行为数据、产品信息、城市信息三张表进行 JOIN，并筛选出 click_product_id > -1 的记录。将查询结果命名为 t1。
    |接着使用用户自定义的 UDAF（CityRemarkUDAF），实现了对区域内商品点击数量的聚合，并生成城市备注信息。将聚合结果命名为 t2。
    |然后使用 Spark SQL 的窗口函数 rank()，在每个区域内对商品的点击数量进行排名。将排名结果命名为 t3。
    |最后从 t3 表中选择排名在前三位的数据，并显示在控制台上
    |
    |""".stripMargin

  def topN(): Unit = {
    // 创建 Spark 运行配置对象
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Hello")
    //创建 SparkSession 对象
    val spark: SparkSession = SparkSession.builder().config(sparkConf)
      .enableHiveSupport() // 添加Hive支持
      .getOrCreate()
    spark.sql("use spark")
    // 查询基础数据
    spark.sql(
      """
        |select
        |a.*,
        |p.product_name,
        |c.area,
        |c.city_name
        |from user_visit_action a
        |join product_info p on a.click_product_id = p.product_id
        |join city_info c on a.city_id = c.city_id
        |where a.click_product_id > -1
        |""".stripMargin
    ).createOrReplaceTempView("t1")
    //根据区域，商品进行数据聚合
    spark.udf.register("cityRemark", functions.udaf(new CityRemarkUDAF()))
    spark.sql(
      """
        |
        |select
        |area,
        |product_name,
        |count(*) as clickCnt,
        |cityRemark(city_name) as city_remark
        |from t1
        |group by area, product_name
        |""".stripMargin
    ).createOrReplaceTempView("t2")
    spark.sql(
      """
        |select
        |*,
        |rank() over(partition by area order by clickCnt desc) as rank
        |from t2
        |""".stripMargin
    ).createOrReplaceTempView("t3")
    spark.sql(
      """
        |select
        |*
        |from t3
        |where rank <=3
        |""".stripMargin
    ).show(false)
    spark.stop()

  }

  """
    |CityRemarkUDAF 用于计算城市的点击数量并生成城市备注信息
    |""".stripMargin

  case class Buffer(var total: Long, var cityMap: mutable.Map[String, Long])
  """
    | 自定义聚合函数：实现城市备注功能
    | 1. 继承Aggregator, 定义泛型
    | IN ： 城市名称
    | BUF : Buffer =>【总点击数量，Map[（city, cnt）, (city, cnt)]】
    | OUT : 备注信息
    | 2. 重写6个方法
    |""".stripMargin

  class CityRemarkUDAF extends Aggregator[String, Buffer, String] {

    override def zero: Buffer = {
      Buffer(0, mutable.Map[String, Long]())
    }

    override def reduce(buff: Buffer, city: String): Buffer = {
      buff.total += 1
      val newCount = buff.cityMap.getOrElse(city, 0L) + 1
      buff.cityMap.update(city, newCount)
      buff
    }

    override def merge(b1: Buffer, b2: Buffer): Buffer = {
      b1.total += b2.total
      val map1 = b1.cityMap
      val map2 = b2.cityMap
      map2.foreach {
        case (city, cnt) => {
          val newCount = map1.getOrElse(city, 0L) + cnt
          map1.update(city, newCount)
        }
      }
      b1.cityMap = map1
      b1
    }

    // 将统计的结果生成字符串信息
    override def finish(reduction: Buffer): String = {
      val remarkList = ListBuffer[String]()
      val totalcnt = reduction.total
      val cityMap = reduction.cityMap
      // desc order
      val cityCntList = cityMap.toList.sortWith(
        (left, right) => {
          left._2 > right._2
        }
      ).take(2)
      val hasMore = cityMap.size > 2
      var rsum = 0L
      cityCntList.foreach {
        case (city, cnt) => {
          val r = cnt * 100 / totalcnt
          remarkList.append(s"${city} ${r}%")
          rsum += r
        }
      }
      if (hasMore) {
        remarkList.append(s"其他 ${100 - rsum}%")
      }

      remarkList.mkString(", ")
    }

    // Spark 用于序列化缓冲区的编码器
    override def bufferEncoder: Encoder[Buffer] = Encoders.product

    // Spark 用于序列化输出结果的编码器
    override def outputEncoder: Encoder[String] = Encoders.STRING
  }
}
